package com.lvmama.rhino.analyze.client

import com.lvmama.rhino.analyze.client.MonitorApi.logger
import com.lvmama.rhino.analyze.processing._
import com.lvmama.rhino.common.entity.{JobConfig, PerformaceMonitorLog}
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.logging.LazyLogging
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions.col
import org.elasticsearch.spark.sql._
import org.apache.spark.storage.StorageLevel

/**
  * Created by Administrator.
  */
class MonitorApi (config: JobConfig, path: String) extends SparkApplication  with LazyLogging {
  override var appName: String = "MonitorApi"
  override var sparkConfig: Map[String, String] = config.spark

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)
    sparkConfig += ("spark.master" -> "local[3]")
    withSparkContext { sc =>
      val sqlCtx = new SQLContext(sc)
      import sqlCtx.implicits._
      //预处理
      val yesterday = Utils.getYesterdayTimeInMillis()
      val preLog = sc.textFile(path).map(l => PerformanceMonitor.prePerformanceMonitor(l))
      val log =  preLog.flatMap{
        x =>x._15.map(y =>PerformaceMonitorLog.fullProcess(x._1, x._2, x._3, x._4, x._5, x._6, x._7, x._8, x._9, x._10, x._11, x._12, x._13, x._14, y, yesterday))
      }.toDF()
      val result = log.filter(!col("apiName").equalTo(""))

      val data = result.select(col("operators"),col("networkType"),col("deviceToken"),col("tec"),col("version"),col("province"),col("city"),
        col("mobileNumber"),col("userId").as("usersId"),col("networkSpeed"),col("osVersion"),col("browser"),col("deviceName"),col("interfaceUniqueFlag"),col("pageId"),
        col("networkTime"),col("apiTime"),col("apiName"),col("apiVersion"),col("requestType"),col("time"),col("errorCode"))
        .dropDuplicates(Seq("interfaceUniqueFlag"))

      logger.info("monitorApi count:"+data.count())
      val today = Utils.getToday("yyyyMMdd")
      logger.info("monitorApi index:" + today)
      data.saveToEs(today+"/monitorApi",Map("es.mapping.id" -> "interfaceUniqueFlag"))
      logger.info("monitorApi after count:"+data.count())
    }
  }
}

object MonitorApi  extends LazyLogging {
  def apply(path: String): Unit = {
    logger.info("monitorApi path:" + path)
    val config = JobConfig()
    new MonitorApi(config, path).execute()
  }
}